package com.yjd.bi.cannal;

import java.util.Date;
import java.util.List;

import org.apache.commons.lang.SystemUtils;
import org.apache.log4j.Logger;
import org.springframework.util.Assert;

import com.alibaba.fastjson.JSON;
import com.alibaba.otter.canal.client.CanalConnector;
import com.alibaba.otter.canal.protocol.CanalEntry.Column;
import com.alibaba.otter.canal.protocol.CanalEntry.Entry;
import com.alibaba.otter.canal.protocol.CanalEntry.EntryType;
import com.alibaba.otter.canal.protocol.CanalEntry.EventType;
import com.alibaba.otter.canal.protocol.CanalEntry.RowChange;
import com.alibaba.otter.canal.protocol.CanalEntry.RowData;
import com.alibaba.otter.canal.protocol.CanalEntry.TransactionBegin;
import com.alibaba.otter.canal.protocol.CanalEntry.TransactionEnd;
import com.alibaba.otter.canal.protocol.Message;
import com.google.protobuf.Descriptors.FieldDescriptor;
import com.google.protobuf.InvalidProtocolBufferException;
import com.yjd.bi.canal.enums.CanalListenedTables;
import com.yjd.bi.canal.model.CanalMessageModel;
import com.yjd.bi.canal.model.Entrys;
import com.yjd.bi.canal.service.ICanalMessageService;
import com.yjd.bi.canal.util.CanalListenedTableUtil;
import com.yjd.comm.rpm.RpmServiceKeyEnum;
import com.yjd.comm.util.FrameUtil;
import com.yjd.comm.util.ServiceHolder;

public class CannelClient {
	 	protected final static Logger             logger             =  Logger.getLogger(CannelClient.class);
	    protected static final String             SEP                = SystemUtils.LINE_SEPARATOR;
	    protected static final String             DATE_FORMAT        = "yyyy-MM-dd HH:mm:ss";
	    protected volatile boolean                running            = false;
	    protected Thread.UncaughtExceptionHandler handler            = new Thread.UncaughtExceptionHandler() {

	                                                                     public void uncaughtException(Thread t, Throwable e) {
	                                                                         logger.error("parse events has an error", e);
	                                                                     }
	                                                                 };
	    protected Thread                          thread             = null;
	    protected CanalConnector                  connector;
	    protected String                          destination;
	    private static ICanalMessageService canalMessageService;

	    public CannelClient(String destination){
	        this(destination, null);
	    }

	    public CannelClient(String destination, CanalConnector connector ){
	        this.destination = destination;
	        this.connector = connector;
	    }

	    public boolean isWorking(){
	    	return this.thread.isAlive();
	    }
	    
	    protected void start() {
	        Assert.notNull(connector, "connector is null");
	        thread = new Thread(new Runnable() {
	            public void run() {
	                process();
	            }
	        });
	        thread.setUncaughtExceptionHandler(handler);
	        thread.setName(destination);
	        thread.start();
	        running = true;
	    }

	    protected void stop() {
	        if (!running) {
	            return;
	        }
	        running = false;
	        if (thread != null) {
	            try {
	                thread.join();
	            } catch (InterruptedException e) {
	                // ignore
	            }
	        }

	    }

	    protected void process() {
	        int batchSize = 5 * 1024;
	        while (running) {
	            try {
	                connector.connect();
	                connector.subscribe();
	                while (running) {
	                    Message message = connector.getWithoutAck(batchSize); // 获取指定数量的数据
	                    long batchId = message.getId();
	                    if(this.record(message)){
	                    	connector.ack(batchId); // 提交确认
	                    }else{
	                    	connector.rollback(batchId); // 处理失败, 回滚数据
	                    }
	                }
	            } catch (Exception e) {
	                logger.error("process error!", e);
	            } finally {
	                connector.disconnect();
	            }
	        }
	    }
	    
	    public void recordMessage(Message message) throws Exception{
			long batchId = message.getId();
	        int size = message.getEntries().size();
	        List<CanalMessageModel> list=FrameUtil.newArrayList();
	        if (batchId == -1 || size == 0) {
	        } else {
	            for (Entry entry : message.getEntries()) {
		               /**
	            	logger.info(entry.getHeader().getLogfileName()+":"+entry.getHeader().getLogfileOffset());
	            	logger.info(entry.getHeader().getSchemaName()+":"+entry.getHeader().getTableName());
	                * 判断是否是要监控的表
	                */
	                if(!CanalListenedTableUtil.isInListened(entry.getHeader().getTableName(), entry.getHeader().getSchemaName())){
	                	continue;
	                }
	                if (entry.getEntryType() == EntryType.TRANSACTIONBEGIN || entry.getEntryType() == EntryType.TRANSACTIONEND) {
		                continue;
		            }
	                long executeTime = entry.getHeader().getExecuteTime();
 		            long delayTime = new Date().getTime() - executeTime;
		            if (entry.getEntryType() == EntryType.ROWDATA) {
		                RowChange rowChage = null;
		                try {
		                    rowChage = RowChange.parseFrom(entry.getStoreValue());
		                } catch (Exception e) {
		                    throw new RuntimeException("parse event has an error , data:" + entry.toString(), e);
		                }
		                EventType eventType = rowChage.getEventType();
		                if (eventType == EventType.QUERY || rowChage.getIsDdl()) {
		                	 CanalMessageModel q=new CanalMessageModel();
				                q.setDelay( delayTime);
				                q.setEvent_type(eventType.getNumber());
				                q.setEntry_type(entry.getEntryType().getNumber());
			                    q.setEntrydata(rowChage.getSql());
			                    q.setTable_name(entry.getHeader().getTableName());
			                    q.setSchema_name(entry.getHeader().getSchemaName());
					            q.setCreate_time(System.currentTimeMillis());
					            //QUERY记录跳过处理
					            q.setStatus(eventType == EventType.ALTER?0:1);
			                    list.add(q);
			                    continue;
		                }
		                
		                for (RowData rowData : rowChage.getRowDatasList()) {
			                List<Entrys> elist=FrameUtil.newArrayList();
		                	CanalMessageModel m=new CanalMessageModel();
		 	                m.setStatus(0);
		 	                m.setDelay(delayTime);
		 	                m.setEvent_type(eventType.getNumber());
		 		            m.setTable_name(entry.getHeader().getTableName());
		 	                m.setSchema_name(entry.getHeader().getSchemaName());
		 	                m.setBinlogfile(entry.getHeader().getLogfileName());
		 	                m.setBinlogfile_offset(entry.getHeader().getLogfileOffset());
		 		            m.setEntry_type(entry.getEntryType().getNumber());
		 		            m.setExecute_time(executeTime);
		                	for (Column column : rowData.getBeforeColumnsList()) {
		                		Entrys e=new Entrys();
		        	            StringBuilder builder = new StringBuilder();
		        	            builder.append(column.getName() + " : " + column.getValue());
		        	            builder.append("    type=" + column.getMysqlType());
		        	            if (column.getUpdated()) {
		        	                builder.append("    update=" + column.getUpdated());
		        	            }
		        	            e.setField(column.getName());
		        	            e.setValue_old(column.getValue());
		        	            e.setType(column.getMysqlType());
		        	            e.setUpdate(column.getUpdated()?1:0);
		        	            e.setIskey(column.getIsKey()?1:0);
		        	            elist.add(e);
		        	        }
		                	for (Column column : rowData.getAfterColumnsList()) {
		                		Entrys e=new Entrys();
		                		boolean isnew=true;
		                		for(Entrys es:elist){
		        	            	if(es.getField().equals(column.getName())){
		        	            		e=es;
		        	            		isnew=false;
		        	            		break;
		        	            	}
		        	            }
		        	            e.setField(column.getName());
		        	            e.setValue_new(column.getValue());
		        	            e.setType(column.getMysqlType());
		        	            e.setUpdate(column.getUpdated()?1:0);
		        	            e.setIskey(column.getIsKey()?1:0);
		        	            if(isnew)
		        	            	elist.add(e);
		        	        }
		                	for(Entrys e:elist){
			                	if(e.getIskey().intValue()==1){
			                		m.setP_key(e.getValue_new()==null?e.getValue_old():e.getValue_new());
			                		continue;
			                	}
			                }
			                m.setEntrydata(JSON.toJSONString(elist));
				            m.setCreate_time(System.currentTimeMillis());
			                list.add(m);
		                }
		                
		            }
		        }
	        }
	        if(list.size()>0){
	        	getCanalMessageService().inserBatch(list);
	        }
		}

	    private static ICanalMessageService getCanalMessageService(){
	    	if(canalMessageService==null){
	    		canalMessageService=ServiceHolder.getService(ICanalMessageService.class, "canalMessageService", RpmServiceKeyEnum.RPM_BI_SERVICE_KEY_E);
	    	}
	    	return canalMessageService;
	    }
	    
	    public void setConnector(CanalConnector connector) {
	        this.connector = connector;
	    }
	    
	    public  boolean record( Message message){
	    	try {
				recordMessage(message);
			} catch (Exception e) {
				return false;
			}
	    	return true;
	    };
}
